﻿using System;
using System.Collections.Generic;
using System.Collections.Specialized;
using System.Net;
using Rhino.Queues;
using Rhino.Queues.Model;
using SimpleServiceBus.Bus;
using SimpleServiceBus.Bus.Transport;
using SimpleServiceBus.Utilities;
using SimpleServiceBus.Utilities.Network;

namespace SimpleServiceBus.Endpoints.Rhino.Transport
{
    public class RhinoQueuesTransport : BaseTransport
    {
        public const int DefaultRhinoQueuesPort = 2200;

        private QueueManager _queue;

        private readonly IDictionary<Guid, int> _failuresPerMessage = new Dictionary<Guid, int>();

       
        public RhinoQueuesTransport()
        {
            QueueStoragePath = "ssb.queue.storage";
            QueueUri = new Uri("rhino://" + Dns.GetHostName());
            InputQueueName = "queue";
            PoisonQueueName = "poison";
        }

        public string QueueStoragePath { get; set; }
        public Uri QueueUri { get; set; }
        public string InputQueueName { get; set; }
        public string PoisonQueueName { get; set; }

        public override string ReturnAddress
        {
            get
            {
                if (QueueUri != null)
                    return QueueUri.AbsoluteUri + InputQueueName;
                if (Endpoint != null && Endpoint.Metadata != null)
                    return Endpoint.Metadata.ReturnAddress;

                return null;
            }
        }

        protected override void Receive()
        {
            new TransactionWrapper().RunInTransaction(base.Receive);
        }

        public override void Send(MessageEnvelope m, string[] destinations)
        {
            new TransactionWrapper().RunInTransaction(() =>
              {
                  foreach (string destination in destinations)
                  {
                      _queue.Send(new Uri(destination),
                                  new MessagePayload
                                      {
                                          Data = m.Payload,
                                          Headers =
                                              new NameValueCollection(
                                              m.Headers)
                                      });

                  }
              });

        }

        public override int GetNumberOfPendingMessages()
        {
            if (_queue != null)
            {
                var msgs = _queue.GetMessagesCurrentlySending();
                if (msgs != null)
                    return msgs.Length;
            }
            return 0;
        }

        protected override MessageEnvelope GetNextMessage()
        {
            Message m = null;
            MessageEnvelope envelope = null;
            try
            {
                m = _queue.Receive(InputQueueName);
                if (m == null) return null;

                envelope = CreateTransportMessageFromRhinoMessage(m);

                lock (_failuresPerMessage)
                    _failuresPerMessage.Remove(m.Id.MessageIdentifier);

            }
            catch (Exception ex)
            {
                bool shouldThrow;
                HandleMessageReceiveError(m, ex, out shouldThrow);

                if (shouldThrow) throw;
            }

            return envelope;
        }

        protected virtual IPAddress GetIPaddressForQueue(Uri queueUri)
        {
            return queueUri.ToIPAddress();
        }

        protected void BindToInputQueue()
        {
            IPAddress ip = GetIPaddressForQueue(QueueUri);
            int port = DefaultRhinoQueuesPort;

            if (QueueUri.Port > 0)
                port = QueueUri.Port;

            _queue = new QueueManager(new IPEndPoint(ip, port), QueueStoragePath);
            _queue.CreateQueues(InputQueueName, PoisonQueueName);
            
        }

        protected override void StartService()
        {
            BindToInputQueue();

            base.StartService();
        }

        protected override void ShutDownService()
        {
            if (_queue != null)
                _queue.Dispose();

            base.ShutDownService();
        }

        protected virtual MessageEnvelope CreateTransportMessageFromRhinoMessage(Message rhinoMessage)
        {
            MessageEnvelope result = CreateEmptyMessageEnvelope();

            if (rhinoMessage.Headers != null && rhinoMessage.Headers.Count > 0)
            {
                foreach (string key in rhinoMessage.Headers.Keys)
                    result.Headers.Add(key, rhinoMessage.Headers[key]);
            }

            result.Payload = rhinoMessage.Data;
            return result;
        }

        protected virtual void HandleMessageReceiveError(Message message, Exception error, out bool shouldSuppressException)
        {
            //Log error

            if (message != null)
            {
                lock (_failuresPerMessage)
                {
                    if (!_failuresPerMessage.ContainsKey(message.Id.MessageIdentifier))
                        _failuresPerMessage[message.Id.MessageIdentifier] = 1;
                    else
                        _failuresPerMessage[message.Id.MessageIdentifier] = _failuresPerMessage[message.Id.MessageIdentifier] + 1;

                    if (_failuresPerMessage[message.Id.MessageIdentifier] == MaxRetries)
                    {
                        _failuresPerMessage.Remove(message.Id.MessageIdentifier);

                        if (_queue != null)
                            _queue.MoveTo(PoisonQueueName, message);

                        shouldSuppressException = true;
                        return;
                    }
                }
            }
            shouldSuppressException = false;
        }
    }
}
